package com.atguigu.day04;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2024/4/2
 * 该案例演示了富函数
 *      Flink的算子在执行的时候，需要将处理函数作为参数传递给算子。处理函数中完成的就是当前算子要执行的操作
 *      Flink一般是接口的形式，定义的处理函数。例如：MapFunction、FilterFunction、ReduceFunction....
 *      除了接口之外，每一个处理函数都有一个对应的富函数类（抽象类），实现了处理函数接口。命名方式 Rich + 处理函数接口
 *      与上文对应 MapFunction----RichMapFunction FilterFunction---RichFilterFunction
 * 富函数提供了更丰富的功能
 *      可以获取运行时上下文
 *      可以调用算子的生命周期相关的方法，以map算子为例
 *          open:算子初始化时执行的方法, 在每个并行子任务上都会执行一次
 *          map: 每条数据到来的时候都要执行的方法
 *          close:算子执行结束的时候执行的方法, 在每个并行子任务上都会执行一次。
 *                 注意：如果是无界流，close方法一直不会被调用
 * 需求：从指定的网络端口中读取数据，在每条读取的数据前加"atguigu->"前缀
 */
public class Flink01_Rich_Function {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        /*SingleOutputStreamOperator<String> mapDS = socketDS.map(
                new MapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        return "atguigu->" + value;
                    }
                }
        );*/
        SingleOutputStreamOperator<String> mapDS = socketDS.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext runtimeContext = getRuntimeContext();
                        String name = runtimeContext.getTaskNameWithSubtasks();
                        int index = runtimeContext.getIndexOfThisSubtask();
                        System.out.println("~~~编号是"+index+",名字是"+name+"的子任务初始化了~~~");
                    }

                    @Override
                    public void close() throws Exception {
                        System.out.println("~~~close~~~");

                    }

                    @Override
                    public String map(String value) throws Exception {
                        return "atguigu->" + value;
                    }
                }
        );
        mapDS.print();
        env.execute();
    }
}
